use super::{
    device::{CaCertificate, RpcCertificate},
    geoip::GeoIpHandler,
};
use crate::{
    fairing::record::RecordDb,
    ipc::server::{insert_task, ipc_server_task},
};
use rocket::fairing::{Fairing, Info, Kind};
use rocket::tokio;
use rocket::Orbit;
use rocket::Rocket;
use rocket_db_pools::Database;
use tokio::sync::{mpsc, oneshot, Mutex};

struct IpcHandle {
    ipc_handle: tokio::task::JoinHandle<()>,
    insert_handle: tokio::task::JoinHandle<()>,
    stop_tx: oneshot::Sender<()>,
}

pub struct Ipc {
    handle: Mutex<Option<IpcHandle>>,
}

impl Ipc {
    pub fn new() -> Ipc {
        Ipc {
            handle: Mutex::new(None),
        }
    }
}

#[rocket::async_trait]
impl Fairing for Ipc {
    fn info(&self) -> Info {
        Info {
            name: "IPC",
            kind: Kind::Liftoff | Kind::Shutdown | Kind::Singleton,
        }
    }

    async fn on_liftoff(&self, rocket: &Rocket<Orbit>) {
        let (tx, rx) = mpsc::channel(100);

        let ca_cert: &CaCertificate = rocket
            .state()
            .expect("Certificates must be initialized before IPC");
        let ca_cert: std::sync::Arc<super::device::CertificateItem> = ca_cert.0.clone();

        let rpc_cert: &RpcCertificate = rocket
            .state()
            .expect("Certificates must be initialized before IPC");
        let rpc_cert = rpc_cert.0.clone();

        let geoip_reader: &GeoIpHandler = rocket
            .state()
            .expect("GeoIP must be initialized before IPC");
        let geoip_reader = geoip_reader.reader();

        let mut handle = self.handle.lock().await;

        let record_db = RecordDb::fetch(rocket).expect("Record database is not ready");
        let record_db = unsafe {
            let db_ptr: *const RecordDb = record_db;
            let static_db: &'static RecordDb = &*db_ptr;
            static_db
        };

        let ipc_handle = tokio::spawn(async {
            ipc_server_task(record_db, tx, rpc_cert, ca_cert).await;
        });
        let (stop_tx, stop_rx) = oneshot::channel();
        let insert_handle = tokio::spawn(async {
            insert_task(record_db, geoip_reader, rx, stop_rx).await;
        });

        *handle = Some(IpcHandle {
            ipc_handle,
            insert_handle,
            stop_tx,
        });
    }

    async fn on_shutdown(&self, _: &Rocket<Orbit>) {
        let mut handle = self.handle.lock().await;
        let handle = handle.take().unwrap();
        handle.ipc_handle.abort();
        handle
            .ipc_handle
            .await
            .expect_err("IPC server stopped abnormally");
        handle
            .stop_tx
            .send(())
            .expect("Cannot stop database inserting task");
        handle
            .insert_handle
            .await
            .expect("Database inserting task failed");
    }
}
